package util;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Writable;

public class DiffMessage<T extends Parsable<T>> implements Writable {

	private static final Log LOG = LogFactory.getLog(DiffMessage.class);
	public final static String MsgSpt = ":";

	private Class<? extends Parsable> clsT;


	private String mesageBody; // a:b:cls
	private long msgId;

	public DiffMessage() {
		msgId = -1;
		mesageBody = "";
		clsT = null;
	}

	public DiffMessage(long id, String ms) {
		msgId = id;
		mesageBody = ms;
		try {
			T ins = parse(ms).getSecond();
			clsT = ins.getClass();
		} catch (NoSuchMethodException | SecurityException
				| ClassNotFoundException | InstantiationException
				| IllegalAccessException | IllegalArgumentException
				| InvocationTargetException e) {
			e.printStackTrace();
		}
	}

	public DiffMessage(long id, T vla) {
		msgId = id;
		mesageBody = vla.toString();
		clsT = vla.getClass();
	}
	
	public DiffMessage(String idAndMsg, Class<? extends Parsable> cls) {
		if(idAndMsg == null) {
 
			return;
		}
		
		Pair<Long, T> pmg = new Pair<Long, T>();
		try {
			clsT =cls;
			pmg = parse(idAndMsg);
			msgId = pmg.getFirst();
			mesageBody = pmg.getSecond().toString();
		} catch (NoSuchMethodException | SecurityException
				| ClassNotFoundException | InstantiationException
				| IllegalAccessException | IllegalArgumentException
				| InvocationTargetException e) {
			e.printStackTrace();
		}
	
	}
	
	public String getMsg() {
		return mesageBody;
	}
	

	public long getMsgId() {
		return msgId;
	}

	public Pair<Long,T> parse(String msg)
			throws NoSuchMethodException, SecurityException,
			ClassNotFoundException, InstantiationException,
			IllegalAccessException, IllegalArgumentException,
			InvocationTargetException {
		if ( msg != null  ) { //&& (!mesageBody.isEmpty())
			String[] array =msg.split(MsgSpt);
			
			Object objT0 = clsT.newInstance();
			

			Method _parse0 = clsT.getDeclaredMethod("parse",
					new Class[] { String.class });

			if (_parse0 == null ) {
				LOG.error("[DiffMessage] Parameterized Type:"
						+ clsT.getCanonicalName());
				System.out.println("[PRI]ParameterizedType0:"
						+ clsT.getCanonicalName());
				return null;
			}

			T v = (T) _parse0.invoke(objT0, new Object[] { array[1].trim() });

			//System.out.println("parse OK ");

			return new Pair<Long,T>(Long.parseLong(array[0]), v) ;
		}

		return  null;
	}

	@Override
	public void readFields(DataInput in) throws IOException {
//		msgId = in.readLong();
//		mesageBody = in.readUTF();
		String[] msg = in.readUTF().split(MsgSpt);
		msgId = Long.parseLong(msg[0]);
		mesageBody = msg[1];
		
	}

	public void setMsg(String ms) {
		mesageBody = ms;
	}

	public void setMsgId(long id) {
		msgId = id;
	}

	@Override
	public String toString(){
		return msgId + MsgSpt +mesageBody;
		
	}

	@Override
	public void write(DataOutput out) throws IOException {
		//out.writeLong(msgId);
		out.writeUTF(this.toString());
	}

}
